package com.hs.kfk.basic;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description: 基本版本的消息生产者，发送的消息就是一个简单是String
 * @Author 胡尚
 * @Date: 2024/8/7 18:33
 */
public class BasicProducer {
    /**
     * 定义kafka服务端地址
     */
     private final static String BOOTSTRAP_SERVER = "192.168.75.61:9092,192.168.75.62:9092,192.168.75.63:9092";
//    private final static String BOOTSTRAP_SERVER = "worker1:9092,worker2:9092,worker3:9092";
    /**
     * 生产者往哪个topic中发送消息
     */
    private final static String TOPIC_NAME = "disTopic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 设置发送者相关的属性
        Properties properties = new Properties();
        // 设置kafka端口
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        // 配置key的序列化类 去找org.apache.kafka.common.serialization.Serializer接口的实现类
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 配置value的序列化类
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 消息拦截器
        properties.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.hs.kfk.intercepter.MyInterceptor");
        // 创建一个消息生产者对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            // 构建一条消息，  构造方法中的传参是：String topic, K key, V value
            // 这里的key和value的泛型要和上方定义的序列化类型匹配上
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, Integer.toString(i), "MyProducer" + i);

            // 发送消息
            // 发送消息方式一：单向发送，不关心服务端的应答, 仅仅把消息发送给服务器
//            producer.send(record);
            // 发送消息方式二： 同步发送：get()获取服务端应答消息前，会阻塞当前线程。
            RecordMetadata recordMetadata = producer.send(record).get();
            String topic = recordMetadata.topic();
            int partition = recordMetadata.partition();
            long offset = recordMetadata.offset();
            System.out.println("topic:" + topic + "\tpartition:" + partition + "\toffset: " + offset );
            // 发送消息方式三：异步发送：消息发送后不阻塞，服务端有应答后会触发回调函数
//            producer.send(record, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata metadata, Exception exception) {
//                    // 使用 RecordMetadata 对象做相应的操作
//                    if (exception != null){
//                        // 消息发送失败 处理逻辑
//                    }
//                }
//            });
        }
        // 消息生产者 调用close()方法
        producer.close();
    }
}
